package com.atguigu.day03

import com.atguigu.source.{SensorReading, SensorSource}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow

object WindowExample {
  def main(args: Array[String]): Unit = {
    val env:StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val stream: DataStream[SensorReading] = env.addSource(new SensorSource)
    val keyed: KeyedStream[SensorReading, String] = stream.keyBy(_.id)
    val window: WindowedStream[SensorReading, String, TimeWindow] = keyed.timeWindow(Time.seconds(10))
    window.reduce((t1,t2)=>{
      SensorReading(t1.id,0L,t1.temperature.min(t2.temperature))
    }).print()

    env.execute()


  }

}
